package cn.hesion.rocketmqdemo.cn.hesion.consumer;
/**
 * ClassName: MQConsumerConfigure <br/>
 * Description: <br/>
 * date: 2021/11/5 11:02<br/>
 *
 * @author Hesion<br />
 * @version
 * @since JDK 1.8
 */

import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * RocketMQ消费者配置
 * @author: hesion
 * @create: 2021-11-05 11:02
 **/
@Getter
@Setter
@Slf4j
@ToString
@Configuration
@ConfigurationProperties(prefix = "rocketmq.consumer")
public class MQConsumerConfigure {
    private String groupName;
    private String nameSrvAddr;
    private String topics;
    //消费者线程数据量
    private Integer consumeThreadMin;
    private Integer consumeThreadMax;
    private Integer consumeMessageBatchMaxSize;

    @Autowired
    private MQConsumeMsgListenerProcessor consumeMsgListenerProcessor;

    @Bean
    @ConditionalOnProperty(prefix = "rocketmq.consumer", value = "isOnOff",havingValue = "on")
    public DefaultMQPushConsumer defaultMQPushConsumer(){
        log.info("defaultMQPushConsumer 正在创建--------------------------");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(groupName);
        consumer.setNamesrvAddr(nameSrvAddr);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.setMessageListener(consumeMsgListenerProcessor);
        //设置consumer第一次启动从队列头部开始消费还是队列尾部开始消费
        //如果非第一次启动，那么按照上次消费的位置继续消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        //设置消费模型，集群还是广播，默认集群
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //设置一次消费消息的条数，默认1条
        consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);

        try {
            // 设置该消费者订阅的主题和tag，如果订阅该主题下的所有tag，则使用*,
            consumer.subscribe("TestTopic", "*");
            consumer.start();
            log.info("consumer 创建成功 groupName={}, topics={}, namesrvAddr={}", groupName, topics, nameSrvAddr);
        } catch (MQClientException e) {
            log.error("consumer 创建失败!");
        }

        return consumer;
    }
}
